#!usr/bin/env python  
# -*- coding:utf-8 _*-
""" 
@author:robot
@file: tpf_task_v1.py 
@version:
@time: 2022/03/11 
"""
from TapUtil.decorator.some_useful import timeout
from TapUtil.logfunc.logger import logger
from TapClientCore.T04_logic.Dag02.task_body_build.data_copy import DataCopyBuild
from TapClientCore.T04_logic.Dag02.task_body_build.data_sync_common import DataSyncCommon
from TapClientCore.T06_Tapdata_feature.v2.TPF03_task.tpf_task_base import TPFTaskBase
from TapException.TM import *


class V2TPFTask(TPFTaskBase):
    task_type = None
    task_url = None
    TASK_FUNC = {
        'data copy': DataCopyBuild().data_copy,
        'data development': DataSyncCommon().data_sync,
        'data development field rename': DataSyncCommon().data_sync_field_rename,
        'data development add_delete_field': DataSyncCommon().data_sync_add_delete_field,
        'data development left join': DataSyncCommon().data_sync_left_join,
        'data development type modification': DataSyncCommon().data_sync_type_modification,
        'data development field calculation': DataSyncCommon().data_sync_field_calculation,
        'data development one2more': DataSyncCommon().data_sync_merge_one2more,
        'data development more2one': DataSyncCommon().data_sync_merge_more2one,
        'data development one2more2one': DataSyncCommon().data_sync_merge_one2more2one,
        'data development aggregation': DataSyncCommon().data_sync_aggregation,
        'data development rowfilter': DataSyncCommon().data_sync_row_filter,
        'data development javascript': DataSyncCommon().data_sync_javascript,
        'data development custom': DataSyncCommon().data_sync_custom_node
    }

    # 获取字段映射结果
    def get_transform(self, payload):
        return self.TMF_META.get_meta_original_data_by_task_data(payload)

    # 模型推演请求
    def meta_transform(self, payload):
        return self.TMF_TASK.task_meta(payload)

    # 创建task
    def _create_data_copy_development(self, data):
        record = {}
        task_type = self.task_type
        task_config = data.get('task_config')
        task_func = self.TASK_FUNC.get(task_type)
        if task_func is None:
            raise TaskBuildError
        dag = task_func(task_config)
        create_response = None
        if task_type == 'data copy':
            create_response = self.TMF_TASK.create_task_data_copy(dag)
            # 新版本不需要再等待
            # self.TMF_TASK.manage_wait_data_copy_model(create_response.get('name'))
        elif 'data development' in task_type:
            create_response = self.TMF_TASK.create_task_data_development(dag)
        record['id'] = create_response.get('id')
        record['dag'] = dag
        record['create_response'] = create_response
        return record

    # 创建task自动化逻辑
    def create_data_copy_development(self, data):
        self.task_type = data.get('task_type')
        task_name = data.get('task_config').get('config').get('name')
        repeat = self.TMF_TASK.task_check_name_repeat(task_name).get('data')
        # query_result = self.dataflow_query_mongodb({'name': task_name})
        if repeat:
            logger.info("同名Task已存在")
            self.TMF_TASK.task_type = self.task_type
            task_status = self.TMF_TASK.query_task_by_name(task_name).get('status')
            if task_status == 'running':
                self.TMF_TASK.manage_stop_and_wait_stop(task_name)
            try:
                # 删除前重置任务，清除相关slot
                self.TMF_TASK.manage_reset_task(task_name)
                self.TMF_TASK.manage_delete_task(task_name)
                self.task_delete_mongodb({'name': task_name})
            except Exception as e:
                raise e
        logger.info(f"{task_name}创建任务，并将记录持久化")

        # 创建任务
        record = self._create_data_copy_development(data)

        self.task_insert_mongodb(
            {'name': task_name, **record}
        )
        return record

    # 删除任务
    def delete_data_copy_development(self, task_name):
        repeat = self.TMF_TASK.task_check_name_repeat(task_name).get('data')
        if repeat:
            task_status = self.TMF_TASK.query_task_by_name(task_name).get('status')
            if task_status == 'running':
                self.TMF_TASK.manage_stop_and_wait_stop(task_name)
            try:
                self.TMF_TASK.manage_delete_task(task_name)
                self.task_delete_mongodb({'name': task_name})
            except Exception as e:
                assert False

    # 更新任务
    def update_task(self, task_name, update_data):
        self.TMF_TASK.update_data_copy_by_name(task_name, update_data)

    # 更新 处理节点 信息
    def update_task_op_node(self, task_name, node_name, node_update):
        return self.TMF_TASK.update_task_node_by_name(task_name, node_name, node_update)

    # 通过任务名生成console
    def generate_task_url_by_name(self, name, data_type):
        """
        http://192.168.1.181:31025/#/dataflow/details/62394eacdecec63c31d27ad8/
        :param name:
        :return:
        """
        task_id = self.TMF_TASK.query_task_by_name(name).get('id')
        from TapClientCore.T02_tm_api.v2_api.v2_tm import v2tm as TM
        if data_type in ['data copy']:
            from TapClientCore.T05_TM_feature.v2.TMF03_task.tmf_task import TMFSubTask
            self.task_url = f'{TM.serve}/#/migrate/monitor/{task_id}'
        else:
            self.task_url = f'{TM.serve}/#/dataflow/monitor/{task_id}'

    # 获取可观测性中的stats
    def get_task_status(self, name):
        return self.TMF_TASK.query_task_by_name(name).get('status')

    @timeout('GET TASK STATS', 20)
    def _get_task_stats(self, name):
        task_data = self.TMF_TASK.query_task_by_name(name)
        task_id = task_data.get('id')
        task_type = task_data.get('syncType')
        # 2.8版本 copy支持了可观测，
        # dev未支持
        # 2.9版本版本copy和dev都支持可观察性，获取统计事件接口改变
        # 下面的循环不进入
        if task_type == 'sync2.8':
            self.task_new_stats = None
            payload = self.TMF_TASK.get_measurement_payload_task_id(task_id)
            response = self.TMF_TASK.post_measurement_payload(payload)
            if response:
                if response.get('data').get('statistics'):
                    self.task_stats = response.get('data').get('samples')[1]
                    return 'GET TASK STATS'
        else:
            payload = self.TMF_TASK.get_measurement2_payload_task_id(task_id)
            response = self.TMF_TASK.post_measurement2_payload(payload)
            if response:
                if response.get('data').get('samples'):
                    self.task_new_stats = response.get('data').get('samples').get('totalData')
                    return 'GET TASK STATS'

    def get_task_stats(self, name):
        self._get_task_stats(name)
        if self.task_new_stats:
            return self.task_new_stats
        else:
            return self.task_stats

    def get_task_stats_insert(self, name):
        stats = self.get_task_stats(name)
        if self.task_new_stats:
            return stats[0].get('outputInsertTotal')
        else:
            return stats.get('insertedTotal')[0]

    def get_task_stats_update(self, name):
        stats = self.get_task_stats(name)
        if self.task_new_stats:
            return stats[0].get('outputUpdateTotal')
        else:
            return stats.get('updatedTotal')[0]

    def get_task_stats_delete(self, name):
        stats = self.get_task_stats(name)
        if self.task_new_stats:
            return stats[0].get('outputDeleteTotal')
        else:
            return stats.get('deletedTotal')[0]

    # 任务状态调度
    def wait_task_stop(self, name):
        logger.info('WAIT TASK STOP')
        self.TMF_TASK.manage_wait_task_to_stop(name)

    def scheduled_task_wait_it_running(self, name) -> None:
        logger.info('WAIT TASK TO RUNNING')
        response = self.TMF_TASK.query_task_by_name(name)

        self.TMF_TASK.manage_start_and_wait_running(name)

    def scheduled_task_wait_it_stop(self, name):
        self.TMF_TASK.manage_start_and_wait_stop(name)

    def scheduled_task_wait_it_complete(self, name):
        self.TMF_TASK.manage_start_and_wait_complete(name)

    def stop_task_wait_it_paused(self, name) -> None:
        logger.info("STOP AND WAIT TASK TO PAUSED")
        try:
            self.TMF_TASK.manage_stop_and_wait_stop(name)
        except Exception as e:
            self.TMF_TASK.manage_force_stop(name)

            class StopTimeOut(Exception):
                ...

            raise StopTimeOut()

    def task_is_running(self, task_name):
        task_status = self.TMF_TASK.query_task_by_name(task_name).get('status')
        if task_status == 'running':
            return True
        return False

    # 获取所有日志
    def get_subtask_info_log_all(self, task_id):
        raw_log = self.TMF_TASK.get_subtask_log(task_id)
        log_list = [f'{k}' for k in raw_log]
        log_string = '\n\n'.join(log_list)
        return log_string

    # 获取任务日志
    def get_subtask_info_by_level(self, task_id, level='ERROR'):
        raw_log = self.TMF_TASK.get_subtask_log(task_id)
        tmp = []
        for one_record in raw_log:
            if one_record.get('level') == level:
                tmp.append(one_record)
        log_list = [f'{k}' for k in tmp]
        log_string = '\n'.join(log_list)
        return log_string
